package com.hdu.dwh.utils;

public class FlinkTypeToPaimonType {
    /**
     * * 将Flink类型转换为Paimon类型
     * @param flinkType
     * @return
     */
    public static org.apache.paimon.types.DataType convertFlinkTypeToPaimonType(org.apache.flink.table.types.DataType flinkType) {
        if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.IntType) {
            return new org.apache.paimon.types.IntType();
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.BigIntType) {
            return new org.apache.paimon.types.BigIntType();
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.DoubleType) {
            return new org.apache.paimon.types.DoubleType();
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.FloatType) {
            return new org.apache.paimon.types.FloatType();
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.BooleanType) {
            return new org.apache.paimon.types.BooleanType();
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.TinyIntType) {
            return new org.apache.paimon.types.TinyIntType();
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.VarCharType) {
            return new org.apache.paimon.types.VarCharType(((org.apache.flink.table.types.logical.VarCharType) flinkType.getLogicalType()).getLength());
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.BinaryType) {
            return new org.apache.paimon.types.BinaryType(((org.apache.flink.table.types.logical.BinaryType) flinkType.getLogicalType()).getLength());
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.DateType) {
            return new org.apache.paimon.types.DateType();
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.TimeType) {
            return new org.apache.paimon.types.TimeType(((org.apache.flink.table.types.logical.TimeType) flinkType.getLogicalType()).getPrecision());
        }
        else if(flinkType.getLogicalType() instanceof org.apache.flink.table.types.logical.TimestampType) {
            return new org.apache.paimon.types.TimestampType(((org.apache.flink.table.types.logical.TimestampType) flinkType.getLogicalType()).getPrecision());
        }
        else {
            throw new UnsupportedOperationException("Unsupported Flink type: " + flinkType.getLogicalType());
        }
    }
}
